package com.xrui;

import com.xrui.config.ConnectionDetails;
import com.xrui.security.Kerboros;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import java.io.IOException;

public class HbaseAccessor {
    public static final byte[] CF = "cf".getBytes();
    public static final byte[] ATTR = "~".getBytes();

    public static void main(String args[]){
        ConnectionDetails connDetails = new ConnectionDetails("Thebe");
        Configuration conf = connDetails.createConfig();

        // Deep copy the configuration.
        Configuration mConf = new Configuration(conf);

        String[] zooKeeperQuorum = connDetails.getZookeeper().getHost().split(",");
        String zooKeeperPort     = connDetails.getZookeeper().getPort();
        // Configure the ZooKeeper quorum:
        mConf.setStrings("hbase.zookeeper.quorum",zooKeeperQuorum );
        mConf.setInt("hbase.zookeeper.property.clientPort", Integer.valueOf(zooKeeperPort));

        Kerboros.login(mConf);
        Connection conn = null;
        try {
            conn = ConnectionFactory.createConnection(mConf);
            Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,
                    new SubstringComparator( "0x0002a00018298d1e" )
            );
            HbaseAccessor.QueryByFilter(conn,
                    "ecpcisqeddscat:ecp_INSTRUMENTKEYMASTER_KeyMaster",
                    filter
                    );
        }catch(IOException ioe){

        }finally{
            if(conn != null){
                try {
                    conn.close();
                }catch(IOException ioe){
                    ioe.printStackTrace();
                }
            }
        }
    }

    public static void QueryByFilter(Connection conn,
                                        String tableName,
                                        Filter filter) throws IOException{
        Table table = null ;
        ResultScanner rs = null;

        try {
            table = conn.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.addColumn(CF, ATTR);

            scan.setFilter(filter);

            rs    = table.getScanner(scan);
            for (Result r = rs.next(); r != null; r = rs.next()) {
                byte[] rowKeyByteArr = r.getRow();
                String rowKeyStr =  Bytes.toString(rowKeyByteArr);
                System.out.println(rowKeyStr);
                byte[] cell = null;

            }
        } finally {
            if(rs != null)
                rs.close();
            if( table != null)
                table.close();
        }
    }
}
